package com.hl.rpc

import java.util.UUID
import java.util.concurrent.TimeUnit

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
import scala.concurrent.duration._

/**
  * Created by yuanhailong on 2021/10/8.
  */
class Worker(val masterHost:String,val masterPort:Int,val memory:Int,val cores:Int) extends Actor{


  var masterRef:ActorSelection=_


  private val workderId = UUID.randomUUID().toString

  /**
    * 建立和master的连接
    */
  override def preStart(): Unit ={
    // ip:port ActorSystem acotr

    masterRef = context.actorSelection(s"akka.tcp://MasterSystem@${masterHost}:${masterPort}/user/Master")

    //worker节点向Master 注册消息信息
    //例如Spark的worker节点向Master注册，需要告知WokerId是谁，以及资源情况（memory,cores,etc）
    masterRef ! RegisterWorker(workderId,memory,cores)
  }

  override def receive: Receive = {
    //Worker接收到master确认祖册成功的消息
    case RegisteredWorker(masterURL)=>{

      println(s"""masterURL:${masterURL}""")

      //导入隐式转换
      import context.dispatcher
      //Worker先给自己发送一个心跳消息，然后再通过SelfSendHeart类型消息向Master发送心跳信息
      //定时给自己发，从而发送心跳也是定时的
      context.system.scheduler.schedule(0 millis,Constants.HEARTBEAT_INTERVAL millis,self,SelfSendHeart)
    }

    //发送心跳
    case SelfSendHeart=>{
      println("send heart to master")
      masterRef ! HeartBeat(workderId)
    }
  }

}

object Worker{
  def main(args: Array[String]): Unit = {
    val host=args(0)
    val port=args(1).toInt

    val masterHost=args(2)
    val masterPort=args(3).toInt

    val memory=args(4).toInt

    val cores=args(5).toInt

    val configStr=
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname="${host}"
         |akka.remote.netty.tcp.port="${port}"
       """.stripMargin

    val conf:Config=ConfigFactory.parseString(configStr)

    //获取ActorSystem,他创建并监控下面的Actor
    val actorSystem: ActorSystem = ActorSystem("WorkerSystem",conf)

    //创建Actor
    val worker = actorSystem.actorOf(Props(new Worker(masterHost,masterPort,memory,cores)),"Worker")

    actorSystem.awaitTermination()
  }
}


